package com.example.topN;

import com.example.model.WaterSensor;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;

/**
 * Created with IntelliJ IDEA.
 * ClassName: ProcessAllWindowDemo
 * Package: com.example.topN
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-07-25
 * Time: 19:53
 */

//TopN问题
public class ProcessAllWindowDemo {
    //统计最近10秒钟内出现次数最多的两个水位，并且每5秒钟更新一次
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> data = env.socketTextStream("hadoop102", 7777)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        final String[] s = value.split(" ");
                        return new WaterSensor(s[0], Long.valueOf(s[1]), Integer.valueOf(s[2]));
                    }
                }).assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                //设置升序
                                .<WaterSensor>forMonotonousTimestamps()
                                //从数据中提取水位线
                                .withTimestampAssigner((val, ts) -> val.getTs() * 1000L)
                );

        //思路一：一种最简单的想法是，我们干脆不区分不同水位，
        //而是将所有访问数据都收集起来，统一进行统计计算。所以可以不做keyBy，
        //直接基于DataStream开窗，然后使用全窗口函数ProcessAllWindowFunction来进行处理

        //开窗 设置开窗策略 和 窗口函数
        //滑动窗口大小10秒  步长5秒
        SingleOutputStreamOperator<String> process =
                data.windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                        .process(new MyProcess());

        process.print();
        env.execute();

    }

    public static class MyProcess extends ProcessAllWindowFunction<WaterSensor, String, TimeWindow> {

        @Override
        public void process(Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
            HashMap<Integer, Integer> map = new HashMap<>();

            //1.遍历数据
            for (WaterSensor element : elements) {
                Integer vc = element.getVc();
                //如果不是第一条数据 判断key是否存在
                if (map.containsKey(vc)) {
                    //存在 直接累加
                    //加一操作是取出来之后加一
                    map.put(vc, map.get(vc) + 1);
                } else {
                    //如果key不存在 那么就初始化
                    //第一条数据 赋值为1
                    map.put(vc, 1);
                }


            }

            //对 count进行排序 利用list来实现
            //map 转 list hashMap映射到 Tuple2
            ArrayList<Tuple2<Integer, Integer>> list = new ArrayList<>();
            //遍历hashMap 赋值到list 获取map的Key
            for (Integer vcCount : map.keySet()) {
                list.add(Tuple2.of(vcCount, map.get(vcCount)));
            }

            //降序
            list.sort(new Comparator<Tuple2<Integer, Integer>>() {
                @Override
                public int compare(Tuple2<Integer, Integer> o1, Tuple2<Integer, Integer> o2) {
                    //降序 后 减 前
                    return o2.f1 - o1.f1;
                }
            });

            //3.取出count最大的2个vc
            StringBuffer outStr = new StringBuffer();
            //排序 排序后的list 取最前面的两个值
            for (int i = 0; i < Math.min(2, list.size()); i++) {
                Tuple2<Integer, Integer> data = list.get(i);
                outStr.append("=====================");
                outStr.append("Top" + (i + 1));
                outStr.append("=====================");
                outStr.append("\n");
                outStr.append("vc = " + data.f0);
                outStr.append("\n");
                outStr.append("count = " + data.f1);
                outStr.append("\n");
                outStr.append("窗口结束时间 = " + DateFormatUtils.format(context.window().getEnd(), "yyyy-MM-dd HH:mm:ss"));
                outStr.append("\n");
            }

            out.collect(String.valueOf(outStr));
        }
    }
}
